package com.yl.test;

import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.yl.flink.processor.MysqlCdcParseFunc;
import com.yl.util.JdkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;

/**
 * @author wlf
 * @since 2022/8/8
 * <p>
 * 【mysql配置】
 * 1.开启 binlog
 * server_id=1
 * log_bin=mysql-bin
 * binlog_format=ROW
 * expire_logs_days=1
 * 2.赋权
 * SELECT 允许从表中查看数据
 * REPLICATION SLAVE 允许执行show master status,show slave status,show binary logs命令
 * REPLICATION CLIENT 允许slave主机通过此用户连接master以便建立主从复制关系
 * 3.enable checkpoint
 */
public class FlinkMysqlCdcExample {

    public static void main(String[] args) throws Exception {
        JdkUtil.disableWarning();

        // 从保存点恢复
        Configuration configuration = new Configuration();
//        configuration.setString("execution.savepoint.path", "file:///tmp/flinkcdc/28b1bf7fc1908e83a2578acbd6aaa2c5/chk-12");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);

        // 设置保存点
        env.enableCheckpointing(3000);
        // 设置本地保存（如果是windows默认盘为当前项目所在的盘）
//        env.getCheckpointConfig().setCheckpointStorage("file:///tmp/flinkcdc/");

        // 数据源
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .port(3306)
                // 开启支持新增表
//                .scanNewlyAddedTableEnabled(true)

                // 生产环境
//                .hostname("mysql-prod")
//                .databaseList("db_jyy_auto_report")
//                .tableList("db_jyy_auto_report.methon_dismiss,db_jyy_auto_report.info_fy_targets")
//                .username("support")
//                .password("0?FPS4Cw(WjRnJp1M&=w")

                // 开发环境
                .hostname("elephant")
                .databaseList("db_common")
                .tableList("db_common.data_batch_integrity")
                .username("root")
                .password("MySQL1234!")
                .startupOptions(StartupOptions.initial())
                .deserializer(new JsonDebeziumDeserializationSchema())
                .build();


        env
                .fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "mysql-cdc-source")
                .setParallelism(3)
//                .process(new MysqlCdcParseFunc())
                .print("===>")
                .setParallelism(1);

        env.execute("mysql-cdc-job");
    }
}
